package com.li.flink.app

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson2.JSONObject
import com.li.flink.domain.Access
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
import org.apache.http.client.methods.{CloseableHttpResponse, HttpGet}
import org.apache.http.impl.client.{CloseableHttpClient, HttpClients}
import org.apache.http.util.EntityUtils

object ProvinceUserCntV1 {
  def main(args: Array[String]): Unit = {
    val inputPath = "/Users/lijiacen/Downloads/access.json"
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build()

    val textStream = env.readTextFile(inputPath)
    val mapStream: DataStream[Access] = textStream.map(m => {
      try {
        val obj: Access = JSON.parseObject(m, classOf[Access])
        obj
      } catch {
        case ex: Exception => print("json数据解析失败：" + m)
          null
      }
    })
    val filterStream = mapStream.filter(m => {
      m.getEvent.equals("startup")
    })

    var httpClient: CloseableHttpClient = null
    //增强的并行RichParallelSourceFunction
    var sfMap = new RichMapFunction[Access, Access] {

      override def close(): Unit = {
        if (httpClient != null) {
          httpClient.close()
        }
      }

      override def open(parameters: Configuration): Unit = {
        httpClient = HttpClients.createDefault()
      }

      override def map(in: Access): Access = {
        val url = "https://restapi.amap.com/v3/ip?ip=" + in.getIp + "&key=ddcf94dd65acf4598d2863fe36a0dbfa"
        var response: CloseableHttpResponse = null;
        var province = "-"
        var city = "-"
        try {
          val httpGet = new HttpGet(url)
          response = httpClient.execute(httpGet);
          var code = response.getStatusLine.getStatusCode
          if (code == 200) {
            var entity = response.getEntity
            var result = EntityUtils.toString(entity)
            var jsonObj = JSONObject.parseObject(result)
            province = jsonObj.getString("province")
            city = jsonObj.getString("city")
            in.setProvince(province)
            in.setCity(city)
          }
          in
        } catch {
          case ex: Exception => print("ip转换失败：")
            in
        }
      }
    }
    val keySelector = new KeySelector[Tuple3[String, Int, Int], Tuple2[String, Int]] {
      override def getKey(in: Tuple3[String, Int, Int]): (String, Int) = {
        (in._1, in._2)
      }
    }
    filterStream.map(sfMap).map(m => {
      (m.getProvince, m.getNu, 1)
    }).keyBy(keySelector).sum(2).addSink(new RedisSink[(String, Int, Int)](conf, new RedisProvinceMapper))

    env.execute("ProvinceUserCntV1")
  }
}

class RedisProvinceMapper extends RedisMapper[(String, Int, Int)] {
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET, "province-user_cnt")
  }

  override def getKeyFromData(data: (String, Int, Int)): String = data._1 + "_" + data._2

  override def getValueFromData(data: (String, Int, Int)): String = data._3 + ""
}